package com.charles.bigdata.mapreduce.counter;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordAndLineCounter {

	/**
	 * mapreduce全局计数器,是用枚举类进行计数器的定义 COUNT_LINES用来统计总行数， COUNT_WORDS用来统计总单词数
	 */
	enum MyCouterWordCount {
		COUNT_WORDS, COUNT_LINES
	}

	// 驱动程序
	public static void main(String[] args) throws Exception {

		// 指定hdfs相关的参数
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		// 设置jar包所在路径
		job.setJarByClass(WordAndLineCounter.class);

		job.setMapperClass(WordAndLineCounterMapper.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(LongWritable.class);

		// 本地路径
		Path inputPath = new Path("E:\\temp\\counter.txt");
		FileInputFormat.setInputPaths(job, inputPath);

		Path outputPath = new Path("E:\\temp\\counter");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
		FileOutputFormat.setOutputPath(job, outputPath);

		// 最后提交任务
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}

	private static class WordAndLineCounterMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

			// 获取计数器，行数计数器
			Counter counter = context.getCounter(MyCouterWordCount.COUNT_LINES);
			// 计数器计数
			counter.increment(1L);

			String[] words = value.toString().split(" ");
			for (String word : words) {
				if (!StringUtils.isBlank(word)) {
					context.getCounter(MyCouterWordCount.COUNT_WORDS).increment(1L);
				}
			}

		}
	}
}